#ifndef __WORKER_H
#define __WORKER_H

#define MAX_CLIENTS 1000
#define LISTEN 0
#define ACCEPT 1
#define CONNECT 2
#define SEND 3
#define RECEIVE 4

#include "connection.h"
#include "port.h"

#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

/***************************************/
// Global variables, shared among threads

// Semaphores for synchronization
extern pthread_mutex_t** socket_semaphore;
extern pthread_mutex_t** cq_semaphore;
extern pthread_mutex_t** portmutex;

// Read only
extern HashTable<int> connection_map;
extern HashTable<int> sock2port;
extern int* maxfd, maxID;

// Access to this one need not be synced since
// each thread will only work on its portion of
// connections
extern HashTable<Connection> connections;
extern Heap<Exchange>** connexchanges;

// Access to these needs to be loosely synced
extern Heap<Exchange>** dataexchanges;
extern Heap<Port>* worker_ports;
extern HashTable<Port> ports;

// Must sync access to this
extern fd_set* sockets;

// No sync needed here
extern int* condone, *conorig, total;
extern double* diff, *maxdiff;
extern double maxtime;
extern int* stalled;

/****************************************/

class Worker
{
  int ID;
  fd_set* data_sockets;
  double starttime;
  int succeeded, failed;

 public:
  Worker();
  Worker(int);
  ~Worker();
  int Listen(const char*);
  void Accept();
  void Connect();
  void Run();
};

Worker::Worker()
{
  ID = 0;
  succeeded = failed = 0;
}

Worker::Worker(int id)
{
  ID = id;
  succeeded = failed = 0;
}

Worker::~Worker()
{
  cout << "Worker "<<ID<<" exited ";
}

void Worker::Connect()
{
  int cID = ID % MAXWORKERS;
  // See if we can help someone else
  //if (!connexchanges[ID]->HasMore())
  //{
  //  cID = rand() % MAXWORKERS;
      // cout << "Worker "<<ID<<" will help "<<cID<<endl;
  // }
  //else
  //cID = ID;
  if (!connexchanges[cID]->HasMore())
    {
      if (succeeded > 0 || failed > 0)
	{
	  cout << "worker "<<ID<<" succeeded "<<succeeded<<" failed "<< failed << endl;
	  succeeded = failed = 0;
	}
      return;
     }
  struct timeval tv;
  gettimeofday(&tv, NULL);
  double timenow = tv.tv_sec+tv.tv_usec/1000000.00 - starttime;
  
  Exchange e = connexchanges[cID]->peak();
  if (e.time < timenow)
    {
      pthread_mutex_lock(cq_semaphore[cID]);
      e = connexchanges[cID]->pop();
      pthread_mutex_unlock(cq_semaphore[cID]);
      // Double check, perhaps things have changed
      if (e.time < timenow)
	{
	  Connection* c = e.c;
	  cout << "Worker "<<ID<<" queue "<<cID<<" Trying to connect for "<<c->ID<<" time now "<<timenow<<" should be "<<e.time<<" left connections "<<connexchanges[cID]->Size()<<endl;
	  diff[cID] = diff[cID]+timenow-e.time;
	  if (timenow - e.time > maxdiff[cID])
	    {
	      maxdiff[cID] = timenow - e.time;
	      cout << "Worker "<<ID<<" queue "<<cID<<" maxdiff changed to "<<maxdiff[cID]<<endl;
	    }
	  int sockfd = c->Connect();
	  condone[cID] = condone[cID] + 1;
	  if (!sockfd)
	    {
	      cout << time(0)<<" Could not connect for connection "<< c->ID << endl;
	      failed = failed + 1;
	      c->Close();
	    }
	  else
	    {
	      char msg[MAXL];
	      sprintf(msg,"%d\0",c->peerID);
	      cout << time(0)<<" Successfully connected for connection "<<c->ID << endl;
	      int n = send(sockfd, msg, strlen(msg), 0);
	      if (n == strlen(msg))
		{
		  succeeded = succeeded + 1;
		  cout << time(0)<<" Successfully sent msg for connection "<<c->ID << endl;
		}
	      else
		failed = failed + 1;
	      c->Close();
	    }
	}
    }
  else
    {
      // cout <<"Worker "<<ID<<" queue "<<cID<<" time now "<<timenow<<" next conn at "<<e.time<<" conn ID "<<e.c->ID<<endl;
    }
}

void Worker::Accept()
{
  fd_set temp;
  struct timeval tv;
  tv.tv_sec = 0;
  tv.tv_usec = 1;

  // cout <<"Worker "<<ID<<" waiting on its lock "<<endl;
  pthread_mutex_lock(socket_semaphore[ID]);
  temp = sockets[ID];
  pthread_mutex_unlock(socket_semaphore[ID]);
  //cout <<"Worker "<<ID<<" done with its lock, maxfd "<<maxfd[ID]<<endl;
  select(maxfd[ID]+1, &temp, NULL, NULL, &tv);
  for(int i = 0; i <= maxfd[ID]; i++) 
    {
      if (FD_ISSET(i, &temp))
	{
	  struct sockaddr their_addr;
	  socklen_t addr_size = sizeof(their_addr);
	  struct sockaddr_in* ta;
	  int newfd;
	  newfd = accept(i, (struct sockaddr*) &their_addr, &addr_size);
	  ta = (struct sockaddr_in*)&their_addr;
	  cout << "Worker "<<ID<<" accepted connection from port "<<ntohs(ta->sin_port)<<endl;
	  
	  if (newfd > NONE)
	    {
	      char buf[MAXL];
	      int n = recv(newfd, buf, MAXL, 0);
 	      int gID = ntohs(ta->sin_port)%INC;
	      cout << "Worker "<<ID<<" received "<<n<<" bytes gID "<<gID<<endl;
	      while(gID < maxID)
		{
		  int* port = sock2port.lookup(i);
		  Connection* c = connections.lookup(gID);
		  if (c && c->port == *port && c->sockfd == -1)
		    {
		      cout <<" Found it, ID "<<c->ID<<" guessed "<<gID<<" time "<<c->list[0].time<<endl;
		      break;
		    }
		  gID = gID + INC;
		}
	      if (gID > maxID)
		cout<<"Couldn't find ID "<<buf<<endl;
	      close(newfd);
	      /* 
	      int* port = sock2port.lookup(i);
	      Port* p = ports.lookup(*port);
	      
	      cout <<"Worker "<<ID<<" waiting on port lock "<<(*port)<<endl;
	      pthread_mutex_lock(portmutex[*port]);
	      if (p)
		{
		  p->RemoveConnection();
		  if (!p->GetConnectionCount())
		    close(i);
		}
	      else
		cout << "Couldn't find port for sock "<< i<<endl;
	      pthread_mutex_unlock(portmutex[*port]);
	      cout <<"Worker "<<ID<<" done with port lock "<<endl;	      
	      */
	    }
	}
    }
}

void Worker::Run()
{
  struct timeval tv;
  gettimeofday(&tv, NULL);
  starttime = tv.tv_sec+tv.tv_usec/1000000.00;
  cout << "Worker "<<ID<<" started at "<< starttime << " connections "<<connexchanges[ID]->Size()<<endl;
  condone[ID] = 0;
  conorig[ID] = connexchanges[ID]->Size();
  
  while (TRUE)
    {
      stalled[ID]=stalled[ID]+1;
      gettimeofday(&tv, NULL);
      double timenow = tv.tv_sec+tv.tv_usec/1000000.00;
      //cout << "Worker "<<ID<<" running "<< timenow - starttime << endl;
      Accept();      
      //cout << "Worker "<<ID<<" done with accept "<<timenow-starttime<<endl;
      Connect();
      //cout << "Worker "<<ID<<" done with connect "<<timenow-starttime<<endl;
      usleep(1);
      //if (timenow - starttime > maxtime+ WAIT && !connexchanges[ID]->HasMore())
      //	{
      //  cout << "Worker "<<ID+type*MAXWORKERS<<" exiting"<<endl;
      //  break;
      //}
    }
}

#endif
